package pres.liko.iot.service.impl;

import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.stereotype.Service;
import pres.liko.iot.service.CacheService;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @ClassName: CacheServiceImpl
 * @Auther: liko
 * @Description:
 * @Date: 2022/10/31 21:48
 */
@Service
public class CacheServiceImpl implements CacheService {

    @Autowired
    private ReactiveRedisTemplate reactiveRedisTemplate;

    @Override
    public Mono<Boolean> set(String key, String value) {
        return Flux.just(key, value)
                .all(s -> StringUtils.isBlank(s))
                .flatMap(aBoolean -> reactiveRedisTemplate.opsForValue().set(key, value));
    }

    @Override
    public Mono<String> get(String key) {
        return reactiveRedisTemplate.opsForValue().get(key);
    }

    @Override
    public Mono<Boolean> setValueWithExp(String key, String value, Long expireTime) {
        if (expireTime == 0) {
            return set(key, value);
        }
        return Flux.just(key, value)
                .all(s -> StringUtils.isBlank(s))
                .flatMap(aBoolean -> reactiveRedisTemplate.opsForValue().set(key, value, Duration.ofSeconds(Math.abs(expireTime))));
    }

    @Override
    public Mono<Long> incrementBy(String key, Long step) {
        return Flux.just(key)
                .all(s -> StringUtils.isBlank(s))
                .flatMap(aBoolean -> reactiveRedisTemplate.opsForValue().increment(key, step));
    }

    @Override
    public Mono<Boolean> setNx(String key, String value) {
        return reactiveRedisTemplate.opsForValue().setIfAbsent(key, value);
    }

    @Override
    public Mono<Boolean> hSet(String key, String fieId, String value) {
        return reactiveRedisTemplate.opsForHash().put(key, fieId, value);
    }

    @Override
    public Mono<Boolean> hMset(String key, Map values) {
        return reactiveRedisTemplate.opsForHash().putAll(key, values);
    }

    @Override
    public Mono<Map> hGetAll(String key) {
        Flux<Map.Entry> entries = reactiveRedisTemplate.opsForHash().entries(key);
        return entries.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
    }

    @Override
    public Mono<Long> sAdd(String key, String member) {
        return reactiveRedisTemplate.opsForSet().add(key, member);
    }

    @Override
    public Mono<Long> sAddMembers(String key, Set<String> members) {
        return reactiveRedisTemplate.opsForSet().add(key, members.toArray());
    }

    @Override
    public Mono<Set<String>> sget(String key) {
        return reactiveRedisTemplate.opsForSet().members(key).collect(Collectors.toSet());
    }

    @Override
    public Mono<Long> sdeleteKeys(String... keys) {
        return reactiveRedisTemplate.delete(keys);
    }

    @Override
    public Mono<Boolean> sIsMember(String key, String value) {
        return reactiveRedisTemplate.opsForSet().isMember(key, value);
    }


}
